package cn.itcast.map;

import cn.itcast.avro.SseAvro;
import cn.itcast.avro.SzseAvro;
import cn.itcast.bean.CleanBean;
import org.apache.flink.api.common.functions.RichMapFunction;

import java.math.BigDecimal;

public class UnionMap<T>  extends RichMapFunction<T, CleanBean> {
    @Override
    public CleanBean map(T value1) throws Exception {

       // System.out.println(value1.getClass());
        CleanBean cleanBean = new CleanBean();
        if (value1 instanceof SseAvro){

            final SseAvro value = (SseAvro) value1;
            cleanBean = new CleanBean(
                    value.getMdStreamID().toString(),
                    value.getSecurityID().toString(),
                    value.getSymbol().toString(),
                    value.getTradeVolume(),
                    value.getTotalValueTraded(),
                    BigDecimal.valueOf(value.getPreClosePx()),
                    BigDecimal.valueOf(value.getOpenPrice()),
                    BigDecimal.valueOf(value.getHighPrice()),
                    BigDecimal.valueOf(value.getLowPrice()),
                    BigDecimal.valueOf(value.getTradePrice()),
                    value.getTimestamp(),
                    "sse"
            );
        }else {
            //System.out.println("szse");
            final SzseAvro value = (SzseAvro) value1;
            cleanBean = new CleanBean(
                    value.getMdStreamID().toString(),
                    value.getSecurityID().toString(),
                    value.getSymbol().toString(),
                    value.getTradeVolume(),
                    value.getTotalValueTraded(),
                    BigDecimal.valueOf(value.getPreClosePx()),
                    BigDecimal.valueOf(value.getOpenPrice()),
                    BigDecimal.valueOf(value.getHighPrice()),
                    BigDecimal.valueOf(value.getLowPrice()),
                    BigDecimal.valueOf(value.getTradePrice()),
                    value.getTimestamp(),
                    "szse"
            );
        }

        return cleanBean;
    }
}
